package com.zallds.window;

import com.zallds.utils.Utils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 增量聚合函数
 * AggregateFunction
 * 每经过3秒钟，求安放在北京西站各个红外测温仪所探测到的所有旅客的平均体温
 */
public class InrementAggregateFunctionDemo2 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.62.37", 8889)
                .flatMap(new Utils.Splitter())
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
                .aggregate(new MyAggregateFunction())
                .print();

        env.execute();
    }

    private static class MyAggregateFunction implements AggregateFunction<Tuple2<String, Integer>, Tuple3<String, Integer, Long>, Tuple2<String, Long>> {
        @Override
        public Tuple3<String, Integer, Long> createAccumulator() {
            return new Tuple3<>("", 0, 0L);
        }

        @Override
        public Tuple3<String, Integer, Long> add(Tuple2<String, Integer> value, Tuple3<String, Integer, Long> tuple3) {
            return new Tuple3<>(value.f0, tuple3.f1 + 1, tuple3.f2 + value.f1);
        }

        @Override
        public Tuple3<String, Integer, Long> merge(Tuple3<String, Integer, Long> acc1, Tuple3<String, Integer, Long> acc2) {
            return new Tuple3<>(acc1.f0, acc1.f1 + acc2.f1, acc1.f2 + acc2.f2);
        }

        @Override
        public Tuple2<String, Long> getResult(Tuple3<String, Integer, Long> tuple3) {
            return new Tuple2<>(tuple3.f0, tuple3.f2 / tuple3.f1);
        }
    }
}
